#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time   :2024/11/4 19:25
# @Author :王凯
# @File   :export_xls.py
# @Project:scrapy_spider
import datetime
from pathlib import Path

import pandas as pd
from loguru import logger
from tqdm import tqdm

from apps.pdd.pdd.clean.__dup_data import get_now_mall_ids
from apps.pdd.pdd.clean.main_run import PDDCleanRun
from components.config import WFQ_SOURCE_MYSQL_CONFIG
from utils.db.mysqldb import MysqlDB
import concurrent.futures


class ExportXls(object):

    def __init__(self, version=None):
        if version is None:
            version = datetime.datetime.now().strftime("%Y%m%d")
        self.version = version
        self._to_db = None
        self.shop_info_sql = """
        select version                      as "月份",
               platform_id                  as "平台ID",
               platform_name                as "平台名称",
               mall_id                      as "店铺ID",
               mall_name                    as "店铺名称",
               company_name_src             as "企业名称-店铺表中原始的名称",
               taxpayer_id                  as "统一社会信用代码",
               company_name                 as "企业名称-治理后的最新名称",
               month_sales                  as "月销量（万件）",
               month_sales_money            as "月销售额（万元）",
               total_sales                  as "累计销量（万件）",
               total_sales_money            as "累计销售额（万元）",
               percent_month_sales          as "月销量同比（%）",
               percent_month_sales_money    as "月销售额同比（%）",
               percent_total_sales          as "累计销量同比（%）",
               percent_total_sales_money    as "累计销售额同比（%）",
               goods_num                    as "商品数（个）",
               self_support                 as "是否自营（是、否）",
               mall_url                     as "店铺URL",
               mall_open_time               as "店铺开店时间",
               mall_open_time_year          as "店铺开店时长（年）",
               mall_status                  as "店铺状态",
               mall_end_time                as "店铺关店时间",
               mall_platform_classification as "店铺主营分类（平台分类）",
               mall_business_classification as "店铺主营（商务部分类）",
               mall_scope                   as "店铺经营范围",
               company_province             as "公司所在省份",
               company_city                 as "公司所在城市",
               company_county               as "公司所在区县",
               company_town                 as "公司所在乡镇",
               mall_location_code           as "店铺所在地区代码",
               mall_send_province           as "店铺主要发货省份",
               mall_send_city               as "店铺主要发货城市",
               mall_send_county             as "店铺主要发货区县",
               mall_send_location_code      as "店铺主要发货地区代码",
               mall_desc                    as "店铺简介",
               mall_logo                    as "店铺图片链接",
               mall_star                    as "店铺综合体验评分",
               mall_label                   as "店铺认定资质",
               mall_rating_text             as "店铺服务标签",
               mall_achievements_rank       as "榜单成就排名",
               mall_service_score           as "服务评分",
               mall_details_label_list      as "店铺服务标签"
        from result_pdd_shop_info where mall_id in ({}) and mall_name != ''
        """
        self.good_info_sql = """
        select version as "月份",
        company_name as "公司名称",
        mall_name as "店铺名称",
        mall_id as "店铺ID",
        goods_id as "商品ID",
        goods_name as "商品名称",
        price as "商品价格(元)",
        sale_num as "商品销量",
        level_1 as "一级类目",
        level_2 as "二级类目",
        level_3 as "三级类目",
        goods_rank as "排名",
        hd_url as "商品图片url",
        market_price as "市场价格(元)",
        normal_price as "一般价格(元)",
        tag_list as "商品标签"
        from result_pdd_goods_detail
        where mall_id in ({}) and mall_name != ''
        """

    @property
    def to_db(self):
        if self._to_db is None:
            self._to_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._to_db

    def get_mall_info(self, mall_ids, batch_size=1000):
        datas = []
        for idx, i in enumerate(range(0, len(mall_ids), batch_size)):
            batch_mall_ids_str = ",".join(
                [f"'{i}'" for i in mall_ids[i:i + batch_size]]
            )
            batch_data = self.to_db.find(
                self.shop_info_sql.format(batch_mall_ids_str), to_json=True
            )
            datas.extend(batch_data)

        return pd.DataFrame(datas)

    def get_goods_info(self, mall_ids, batch_size=1000):
        datas = []
        for idx, i in enumerate(range(0, len(mall_ids), batch_size)):
            mall_ids_str = ",".join([f"'{i}'" for i in mall_ids[i:i + batch_size]])
            data = self.to_db.find(
                self.good_info_sql.format(mall_ids_str), to_json=True
            )
            datas.extend(data)
        return pd.DataFrame(datas)

    @logger.catch
    def run_by_ids(self, mall_ids):
        df_mall = self.get_mall_info(mall_ids)
        df_goods = self.get_goods_info(mall_ids)
        return df_mall, df_goods

    def run_mul_by_ids(self, all_mall_ids, max_workers=10, batch_size=10000):
        df_malls = pd.DataFrame()
        df_goods = pd.DataFrame()
        # 加上进度条
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:

            future_to_result = {}

            for i in range(0, len(all_mall_ids), batch_size):
                mall_ids = all_mall_ids[i:i + batch_size]
                future_to_result.update(
                    {executor.submit(self.run_by_ids, mall_ids): mall_ids}
                )

            for future in tqdm(
                concurrent.futures.as_completed(future_to_result),
                total=len(future_to_result),
            ):
                mall_ids = future_to_result[future]
                try:
                    df_mall, df_good = future.result()
                    df_malls = pd.concat([df_malls, df_mall], axis=0)
                    df_goods = pd.concat([df_goods, df_good], axis=0)
                except Exception as e:
                    logger.error(f"{mall_ids} 错误: {e}")

        path = Path(__file__).parent / datetime.datetime.now().strftime("%Y%m%d")
        path.mkdir(parents=True, exist_ok=True)
        df_malls.reset_index(drop=True)
        df_goods.reset_index(drop=True)
        df_malls.sort_values(
            by=["商品数（个）", "店铺开店时长（年）"], ascending=False, inplace=True
        )
        df_malls.drop_duplicates(subset=["店铺ID"], inplace=True, keep="first")
        df_goods.drop_duplicates(subset=["店铺ID", "商品ID"], inplace=True)

        # 每 batch_size 生成一个文件
        # 多线程生成

        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            future_to_result = {}

            for idx, i in enumerate(range(0, len(df_malls), batch_size), start=1):
                batch_mall = df_malls[i:i + batch_size]
                batch_good = df_goods[df_goods["店铺ID"].isin(df_malls["店铺ID"])]
                logger.info(f"{idx} 开始")
                future_to_result.update(
                    {executor.submit(self.gen_xls, batch_mall, batch_good, idx): idx}
                )

            for future in tqdm(
                concurrent.futures.as_completed(future_to_result),
                total=len(future_to_result),
            ):
                idx = future_to_result[future]
                try:
                    future.result()
                except Exception as e:
                    logger.error(f"{idx} 错误: {e}")
                finally:
                    logger.info(f"{idx} 完成")

    def gen_xls(self, df_malls, df_goods, idx):
        path = Path(__file__).parent / datetime.datetime.now().strftime("%Y%m%d")
        path.mkdir(parents=True, exist_ok=True)
        df_malls.to_csv(path / f"店铺信息_{idx}.csv")

        df_malls = df_malls.reset_index(drop=True)
        this_goods_df = df_goods.reset_index(drop=True)

        if this_goods_df.shape[0] > 1048570:
            # 超过最大行数，分批写入
            for idx_j, j in enumerate(
                range(0, this_goods_df.shape[0], 1048570), start=1
            ):
                this_goods_df[j:j + 1048570].reset_index(drop=True).to_csv(
                    path / f"商品详情_{idx}_{idx_j}.csv"
                )
        else:
            df_goods[df_goods["店铺ID"].isin(df_malls["店铺ID"])].reset_index(
                drop=True
            ).to_csv(path / f"商品详情_{idx}.csv")

    def run(self, max_workers=10):
        sql = """SELECT distinct mall_id FROM result_pdd_shop_info  WHERE mall_name != '' and update_time > '2024-11-05' """
        all_mall_ids_dict = self.to_db.find(sql, to_json=True)
        all_mall_ids = [i["mall_id"] for i in all_mall_ids_dict]
        self.run_mul_by_ids(all_mall_ids, max_workers)

    def run_tmp(self):
        df_0 = pd.read_excel("店铺信息_0.xlsx")
        df_1 = pd.read_excel("店铺信息_10000.xlsx")
        df_2 = pd.read_excel("店铺信息_20000.xlsx")
        df_3 = pd.read_excel("店铺信息_30000.xlsx")

        df_all = pd.concat([df_0, df_1, df_2, df_3], axis=0)

        sql = """SELECT distinct mall_id, version FROM result_pdd_shop_info  WHERE mall_name != '' """
        all_mall_ids_dict = self.to_db.find(sql, to_json=True)
        df_all_mall_ids = pd.DataFrame(all_mall_ids_dict)
        df_all_mall_ids["mall_id"] = df_all_mall_ids["mall_id"].astype(int)
        new_df = df_all_mall_ids[
            ~(
                df_all_mall_ids["mall_id"].isin(df_all["店铺id"])
                & (df_all_mall_ids["version"] == "202410")
            )
        ]
        print(new_df.shape[0])
        new_mall_ids = new_df["mall_id"].tolist()
        self.run_mul_by_ids(new_mall_ids)

    def run_tmp_tmp(self, max_workers=10):
        all_mall_ids = get_now_mall_ids()
        self.run_mul_by_ids(all_mall_ids, max_workers)

    def run_tmp_tmp_(self):
        path_ls = Path(__file__).parent / "20241200"

        for path in path_ls.iterdir():
            if "店铺信息" not in path.as_posix():
                continue
            df = pd.read_excel(path)
            mall_ids_ = df[df["店铺状态"].isin(["该店铺已失效", "该店铺正在上传商品"])][
                "店铺ID"
            ].tolist()

            batch_size = 1000
            with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
                tasks = []
                for i in range(0, len(mall_ids_), batch_size):
                    sub_mall_ids = mall_ids_[i:i + batch_size]
                    tasks.append(
                        executor.submit(
                            PDDCleanRun(version="202412").run_by_ids, sub_mall_ids
                        )
                    )

                for task in tqdm(
                    concurrent.futures.as_completed(tasks), total=len(tasks)
                ):
                    if task.exception():
                        raise task.exception()

    def run_tmp_tmp_tmp(self):
        path_ls = Path(__file__).parent / "20241200"
        all_mall_ids = []

        for path in path_ls.iterdir():
            if "店铺信息" not in path.as_posix():
                continue
            df = pd.read_excel(path)
            all_mall_ids.extend(df["店铺ID"].tolist())
        self.run_mul_by_ids(all_mall_ids, 10)

    def change_df(self):
        path_ls = Path(__file__).parent / "20241202"

        def do_work(path_):
            pd.read_csv(path_).to_excel(path_.as_posix().replace(".csv", ".xlsx"))

        with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
            tasks = []
            for path in path_ls.iterdir():
                logger.debug(path)
                tasks.append(executor.submit(do_work, path))
            for task in tqdm(concurrent.futures.as_completed(tasks), total=len(tasks)):
                if task.exception():
                    raise task.exception()


if __name__ == "__main__":
    test = ExportXls()
    test.run()
    # test.run_tmp_tmp()
    # test.run_tmp_tmp_()
    # test.run_tmp_tmp_tmp()
    # test.change_df()
